package com.scheduler.actor;

import com.scheduler.actor.strategy.RoundRobinWithoutLocalhostRouterStrategy;
import com.scheduler.actor.strategy.RouterStrategy;
import lombok.Getter;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Function;

@Getter
public class WorkerActor {

    /**
     * 对应{@link ActorMessage#getActorName()}
     */
    private final String name;

    /**
     * 多机路由规则，默认是轮询
     */
    private RouterStrategy routerStrategy = new RoundRobinWithoutLocalhostRouterStrategy();
    /**
     * k-v
     * k 对应{@link ActorMessage#getActorDispatcher()}
     * v 对应业务注册的回调对象，ActorMessage 为消息体，Boolean 为提交任务结果，提交成功即可
     */
    private final Map<String, Function<ActorMessage, ?>> dispatcherMap = new HashMap<>();
    /**
     * 执行线程池，如果没有线程池则会阻塞 VertX/Netty
     */
    private ThreadPoolExecutor threadPoolExecutor;
    /**
     * 在提交线程池之前可以转换一下，通常可以用来处理traceId等透传问题
     */
    private Function<Runnable, Runnable> runnableWrapper;

    public WorkerActor(String name) {
        Objects.requireNonNull(name);
        this.name = name;
    }

    public WorkerActor setRouterStrategy(RouterStrategy routerStrategy) {
        Objects.requireNonNull(routerStrategy);
        this.routerStrategy = routerStrategy;
        return this;
    }

    public WorkerActor addDispatcher(String dispatcher, Function<ActorMessage, ?> function) {
        Objects.requireNonNull(dispatcher);
        Objects.requireNonNull(function);
        if (dispatcherMap.containsKey(dispatcher)) {
            throw new IllegalArgumentException("重复定义的dispatcher: " + dispatcher);
        }
        dispatcherMap.put(dispatcher, function);
        return this;
    }

    public WorkerActor setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
        Objects.requireNonNull(threadPoolExecutor);
        this.threadPoolExecutor = threadPoolExecutor;
        return this;
    }

    public WorkerActor setRunnableWrapper(Function<Runnable, Runnable> runnableWrapper) {
        Objects.requireNonNull(runnableWrapper);
        this.runnableWrapper = runnableWrapper;
        return this;
    }
}
